package kafka

// 生产者
import (
	"AR/saas/toollib/logger"
	"errors"
	"fmt"
	"github.com/IBM/sarama"
)

func NewProducer() (sarama.SyncProducer, error) {
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll          // 发送完数据需要leader和follow都确认
	config.Producer.Partitioner = sarama.NewRandomPartitioner // 新选出一个partition
	config.Producer.Return.Successes = true

	client, err := sarama.NewSyncProducer([]string{"127.0.0.1:9092"}, config)
	if err != nil {
		fmt.Println("producer closed, err:", err)
		return nil, errors.New("")
	}

	return client, nil
}

func ProducerSend(topic string, msg string) {
	client, err := NewProducer()
	if err != nil {
		return
	}
	proMsg := &sarama.ProducerMessage{}
	proMsg.Topic = topic
	proMsg.Value = sarama.StringEncoder(msg)
	pid, offset, err := client.SendMessage(proMsg)
	if err != nil {
		logger.Error(fmt.Sprintf("%d,%d", pid, offset))
		return
	}
	logger.Debug("Producer send:" + topic + "success!")
}
